package com.sqoop.commons.receive.impl;

import com.sqoop.commons.receive.MsgReceive;
import com.sqoop.core.msg.Msg;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.jms.core.JmsTemplate;

import javax.jms.Destination;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.concurrent.TimeUnit;

/**
 * Created by zw on 2017/11/10.
 */
public class MqMsgReceiver implements Runnable,MsgReceive {
    private static Logger logger = LogManager.getLogger(MqMsgReceiver.class.getName());

    private String name;
    private JmsTemplate jmsTemplate;
    private Destination destination;
    private MqMsgListener mqMsgListener;
    private MacthErrorMsgHandler macthErrorMsgHandler;
    private volatile boolean runningFlag = true;
    private volatile boolean pauseFlag = false;

    @Override
    public Msg receive() {
        Object obj = jmsTemplate.receiveAndConvert(destination);
        if (obj instanceof Msg){
            Msg msg = (Msg)obj;
            return msg;
        }else {
            logger.error("reveive a msg is not  carry forward Msg.class!!!");
        }
        return null;
    }

    @Override
    public void stop() {
        runningFlag = false;
    }

    @Override
    public void pause() {
        pauseFlag = true;
    }


    public  interface MacthErrorMsgHandler{
       public void process(Msg msg);
    }

    public class DefaultMsgHandler implements MacthErrorMsgHandler{
        @Override
        public void process(Msg msg){
            logger.info(msg.toString());
        }
    }
    public  class LogMsgHandler implements  MacthErrorMsgHandler{
        @Override
        public void process(Msg msg){
            FileOutputStream out = null;
            ObjectOutputStream objectOut = null;
            try {
                out = new FileOutputStream("./ErrorMsg.file");
                objectOut = new ObjectOutputStream(out);
                objectOut.writeObject(msg);
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                try {
                    objectOut.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                try {
                    out.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    public MqMsgReceiver(JmsTemplate jmsTemplate, Destination destination, MqMsgListener mqMsgListener) {
        this.jmsTemplate = jmsTemplate;
        this.destination = destination;
        this.mqMsgListener = mqMsgListener;
        runningFlag = true;
        name = jmsTemplate.toString() + destination.toString();
    }

    @Override
    public void run() {
        while (runningFlag){
            if (pauseFlag){
                try {
                    TimeUnit.SECONDS.sleep(1);
                    continue;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            logger.info("msg start receive!!");
            Msg msg = receive();
            mqMsgListener.addMsg(msg,this.name);
        }

    }

}